package com.tencent.oa.fm.cmp.infrastructure.depend.cache;

import com.google.common.collect.Sets;
import com.tencent.oa.fm.cmp.infrastructure.common.Constants;
import com.tencent.oa.fm.cmp.infrastructure.common.context.ServiceContext;
import com.tencent.oa.fm.cmp.infrastructure.common.exception.ApplicationException;
import com.tencent.oa.fm.cmp.infrastructure.common.serializer.BinarySerializer;
import com.tencent.oa.fm.cmp.infrastructure.common.serializer.JsonSerializer;
import com.tencent.oa.fm.jarvis.core.common.App;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPoolConfig;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * 获取对应的数据
 */
@Slf4j
@Service("redisCacheService")
public class RedisClusterCacheServiceImpl implements RedisCacheService {

    @Resource
    private JedisPoolConfig jedisPoolConfig;

    @Value("${redis.cluster.url}")
    private String redisClusterUrl;

    @Value("${redis.cluster.password}")
    private String redisClusterPassword;

    @Value("${redis.cluster.connectionTimeout}")
    private int redisConnectionTimeout;

    @Value("${redis.cluster.soTimeOut}")
    private int redisSoTime;

    private String DEFAULT_SPLIT_REDIS_HOST = ";";

    private String DEFAULT_SPLIT_REDIS_PORT = ":";

    private JedisCluster jedisCluster;

    private static final int DEFAULT_LOCK_EXPIRE_SECONDS = (int) TimeUnit.SECONDS.toMillis(4);

    private static final int DEFAULT_LOCK_ACQUIRE_EXPIRE_SECONDS = (int) TimeUnit.SECONDS.toMillis(3);

    private static final String DEFAULT_LOCK_VAL_KEY = "-lock";

    private static final Long RELEASE_SUCCESS = 1L;


    @PostConstruct
    public void init() {
        if (StringUtils.isBlank(redisClusterUrl)) {
            log.error("the redisCluster url is null, please set the redisCluster properties..");
        }

        String[] hostPorts = redisClusterUrl.split(DEFAULT_SPLIT_REDIS_HOST);

        Set<HostAndPort> hostAndPortSet = Sets.newHashSet();
        Arrays.stream(hostPorts).forEach(s -> {
            String[] hostAndPort = s.split(DEFAULT_SPLIT_REDIS_PORT);
            HostAndPort hostAndPort1 = new HostAndPort(hostAndPort[0], Integer.parseInt(hostAndPort[1]));
            hostAndPortSet.add(hostAndPort1);
        });
        try {
            jedisCluster = new JedisCluster(hostAndPortSet, redisConnectionTimeout, redisSoTime,
                    jedisPoolConfig.getMaxTotal(), redisClusterPassword, jedisPoolConfig);
        } catch (Exception e) {
            log.error("failed to get the redis service, the error is :", e);
        }
    }

    @Override
    public <T> T get(String key, Class<T> clazz, Class... elementsClass) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        String realKey = wrapCacheKey(key);
        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            byte[] valueBytes = jedisCluster.get(keyBytes);
            if (valueBytes == null || valueBytes.length == 0) {
                return null;
            }
            return BinarySerializer.deSerialize(valueBytes, clazz.getClassLoader());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public <T> void set(String key, @NonNull T value, int expireSecond) {
        this.set(key, value, expireSecond, true);
    }

    @Override
    public <T> void setIfAbsent(String key, @NonNull T value, int expireSecond) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        String realKey = wrapCacheKey(key);
        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            long survivor = jedisCluster.ttl(keyBytes);
            if (survivor < 0) {
                byte[] valueBytes = BinarySerializer.serialize(value);
                jedisCluster.setex(keyBytes, expireSecond, valueBytes);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public long del(String key) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        String realKey = wrapCacheKey(key);
        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            return jedisCluster.del(keyBytes);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 0;
    }

    @Override
    public <T> T fetch(String key, Supplier<T> supplier, int expireSecond, Class<T> clazz, Class... elementsClass) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        T value = this.get(key, clazz);
        if (value != null) {
            return value;
        } else {
            String lockVal = tryAcquireDistributeLock(key, DEFAULT_LOCK_EXPIRE_SECONDS, DEFAULT_LOCK_ACQUIRE_EXPIRE_SECONDS);
            if (lockVal == null) {
                return null;
            }
            try {
                T fetchValue = supplier.get();
                if (ObjectUtils.isEmpty(fetchValue)) {
                    return null;
                }
                this.set(key, fetchValue, expireSecond);
                return fetchValue;
            } catch (Exception e) {
                throw ApplicationException.serverInternalError("redis cache service try unlock error:", e);
            } finally {
                releaseDistributedLock(key, lockVal);
            }
        }
    }

    @Override
    public void releaseDistributedLock(String key, String value) {
        if (StringUtils.isBlank(key) || StringUtils.isBlank(value)) {
            return;
        }
        String realKey = wrapCacheKey(key + DEFAULT_LOCK_VAL_KEY);
        //String localKey = String.join(realKey, DEFAULT_LOCK_VAL_KEY);
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        try {
            Object result = jedisCluster.eval(script, Collections.singletonList(realKey), Collections.singletonList(value));
            if (!RELEASE_SUCCESS.equals(result)) {
                throw ApplicationException.serverInternalError("redis release distributed lock error");
            }
        } catch (Exception e) {
            e.printStackTrace();
            //throw ApplicationException.serverInternalError("redis cache service try unlock error:", e);
        }
    }

    @Override
    public String tryAcquireDistributeLock(String key, long expireMillisecond, long acquireMillisecond) {
        String token = UUID.randomUUID().toString();
        String realKey = wrapCacheKey(key + DEFAULT_LOCK_VAL_KEY);
        try {
            long nano = System.nanoTime();
            do {
                log.info("try lock key: " + realKey);
                //将 key 的值设为 operatorId 1成功  0失败
                String ret = jedisCluster.set(realKey, token, "NX", "PX", expireMillisecond);
                if ("OK".equalsIgnoreCase(ret)) {
                    return token;
                }
                String value = jedisCluster.get(realKey);
                if (token.equals(value)) {
                    jedisCluster.pexpire(realKey, expireMillisecond);
                    return token;
                }
                if (acquireMillisecond <= 0) {
                    log.info("key:"+realKey+";tryAcquireDistributeLock break");
                    //没有设置超时时间，直接退出等待
                    break;
                }
                Thread.sleep(300);
            } while ((System.nanoTime() - nano) < TimeUnit.MILLISECONDS.toNanos(acquireMillisecond));
        } catch (Exception e) {
            e.printStackTrace();
            // throw ApplicationException.serverInternalError("redis cache service try lock error:", e);
            log.error("tryAcquireDistributeLock occurs error",e);
        }
        return null;
    }

    @Override
    public String getRedisKey(String key) {
        String realKey = wrapCacheKey(key + DEFAULT_LOCK_VAL_KEY);
        log.info("try lock key: " + realKey);
        String value = jedisCluster.get(realKey);
        if (StringUtils.isNotEmpty(value)) {
            return value;
        }
        return null;
    }

    @Override
    public Long expire(String key, int seconds) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        String realKey = wrapCacheKey(key);
        try {
            byte[] keyBytes = BinarySerializer.serialize(realKey);
            return jedisCluster.expire(keyBytes, seconds);
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        }
        return 0L;
    }


    @Override
    public <T> Long hset(String key, String field, T value) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        String realKey = wrapCacheKey(key);
        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            byte[] fieldBytes = field.getBytes("UTF-8");
            byte[] valueBytes = BinarySerializer.serialize(value);
            return jedisCluster.hset(keyBytes, fieldBytes, valueBytes);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        }
        return 0L;
    }

    @Override
    public <T> T hget(String key, String field, Class<T> tClass, Class... elementsClass) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        String realKey = wrapCacheKey(key);
        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            byte[] fieldBytes = field.getBytes("UTF-8");
            byte[] resultValue = jedisCluster.hget(keyBytes, fieldBytes);
            return BinarySerializer.deSerialize(resultValue, tClass.getClassLoader());
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        } catch (IOException e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        }
        return null;
    }


    private String wrapCacheKey(String key) {
        String tenantId = ServiceContext.current() == null ? Constants.defaultTenantId
                : (ServiceContext.current().getHeader() == null ? Constants.defaultTenantId : ServiceContext.current().getHeader().getTenantId());
        return String.format("%s-%s-%s", App.getCurrent().getAppInfo().getAppId(), tenantId, key);
    }

    public JedisCluster getJedisCluster() {
        return this.jedisCluster;
    }

    @PreDestroy
    public void destroy() {
        try {
            this.jedisCluster.close();
        } catch (Exception e) {
            log.error("failed to close the jedis", e);
        }
    }

    @Override
    public <T> void lpush(String key, T value, int expireSecond) {
        if (StringUtils.isBlank(key)) {
            return;
        }
        String realKey = wrapCacheKey(key);

        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            byte[] valueBytes = BinarySerializer.serialize(value);
            if (expireSecond > 0) {
                jedisCluster.lpush(keyBytes, valueBytes);
                jedisCluster.expire(realKey, expireSecond);
            } else {
                jedisCluster.set(keyBytes, valueBytes);
            }
        } catch (Exception e) {
            log.info("failed to get the keyBytes, the error is :", e);
        }

    }

    @Override
    public <T> T rpop(String key, Class<T> tClass) {
        if (StringUtils.isBlank(key)) {
            return null;
        }
        String realKey = wrapCacheKey(key);

        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            byte[] value = jedisCluster.rpop(keyBytes);
            return BinarySerializer.deSerialize(value, tClass.getClassLoader());
        } catch (Exception e) {
            log.info("failed to get the keyBytes, the error is :", e);
        }
        return null;
    }

    @Override
    public <T> int sadd(String key, T value, int expireSecond) {
        if (StringUtils.isBlank(key)) {
            return 0;
        }
        String realKey = wrapCacheKey(key);
        Long result = 0l;
        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            byte[] valueBytes = BinarySerializer.serialize(value);
            if (expireSecond > 0) {
                result = jedisCluster.sadd(keyBytes, valueBytes);
                jedisCluster.expire(realKey, expireSecond);
            } else {
                result = jedisCluster.sadd(keyBytes, valueBytes);
            }
        } catch (Exception e) {
            log.info("failed to get the keyBytes, the error is :", e);
        }
        return (result == null || result == 0l) ? 0 : 1;
    }

    @Override
    public <T> Set<T> smembers(String key, Class<T> tClass) {
        if (StringUtils.isBlank(key)) {
            return null;
        }
        String realKey = wrapCacheKey(key);

        try {
            byte[] keyBytes = realKey.getBytes("UTF-8");
            Set<byte[]> redisValue = jedisCluster.smembers(keyBytes);
            if (CollectionUtils.isEmpty(redisValue)) {
                return Sets.newHashSet();
            }
            Set<T> result = Sets.newHashSet();
            for (byte[] bytes : redisValue) {
                T value = BinarySerializer.deSerialize(bytes, tClass.getClassLoader());
                result.add(value);
            }
            return result;
        } catch (Exception e) {
            log.info("failed to get the keyBytes, the error is :", e);
        }
        return null;
    }

    @Override
    public long incrBy(String key, int interval) {
        if (StringUtils.isEmpty(key)) {
            return 0;
        }
        String realKey = wrapCacheKey(key);
        try {
            long value = jedisCluster.incrBy(realKey, interval);
            return value;
        } catch (Exception e) {
            log.error("failed to incr data, the key is :{}", key);
            log.error("failed to incr data, the error is :", e);
            return -1;
        }
    }

    @Override
    public <T> void set(String key, T value, int expireSecond, boolean isSerializable) {
        if (StringUtils.isBlank(key)) {
            throw new IllegalArgumentException("key is null");
        }
        String realKey = wrapCacheKey(key);

        try {
            if (isSerializable) {
                byte[] keyBytes = realKey.getBytes("UTF-8");
                byte[] valueBytes = BinarySerializer.serialize(value);
                // String valueJsonStr = serializeValue(value);
                if (expireSecond > 0) {
                    jedisCluster.setex(keyBytes, expireSecond, valueBytes);
                    //jedisCluster.setex(realKey, expireSecond, valueJsonStr);
                } else {
                    //jedisCluster.set(realKey, valueJsonStr);
                    jedisCluster.set(keyBytes, valueBytes);
                }
            } else {
                if (expireSecond > 0) {
                    jedisCluster.setex(realKey, expireSecond, JsonSerializer.serialize(value));
                    //jedisCluster.setex(realKey, expireSecond, valueJsonStr);
                } else {
                    //jedisCluster.set(realKey, valueJsonStr);
                    jedisCluster.set(realKey, JsonSerializer.serialize(value));
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage(),e);
        }
    }
}
